package com.xxk.rocket.star.core.consume;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.xxk.rocket.star.core.context.ConsumeContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.Future;

public class MessageListener implements MessageListenerConcurrently {

  private static final Logger log = LoggerFactory.getLogger(MessageListener.class);

  private ConsumeHandlerImpl consumeHandler;
  private ConsumeContext consumeContext;

  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExts,
                                                  ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    for (MessageExt messageExt : messageExts) {
      ConsumeHandlerRunnable consumeHandlerRunnable = new ConsumeHandlerRunnable();
      consumeHandlerRunnable.setMessageExt(messageExt);
      consumeHandlerRunnable.setConsumeConcurrentlyContext(consumeConcurrentlyContext);
      consumeHandlerRunnable.setConsumeContext(consumeContext);
      try {
        Future<ConsumeResult> future = consumeHandler.addConsumeRunnable(consumeHandlerRunnable);
      } catch (InterruptedException e) {
        log.error("InterruptedException:", e);
      }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }

  public ConsumeHandlerImpl getConsumeHandler() {
    return consumeHandler;
  }

  public void setConsumeHandler(ConsumeHandlerImpl consumeHandler) {
    this.consumeHandler = consumeHandler;
  }

  public ConsumeContext getConsumeContext() {
    return consumeContext;
  }

  public void setConsumeContext(ConsumeContext consumeContext) {
    this.consumeContext = consumeContext;
  }
}
